#include "config.h"
#include "common.h"

DataNode* datanodes=NULL;
long requestid=0;
int** block_loc=NULL;
int conn2client_fd[DATANODE_NUM];
long start_time,end_time,s_time;
long write_total_time=0;
long read_total_time=0;
long total_time=0;
FILE* latency_file=NULL;

void play_trace_msr(char* tracepath);
void play_trace_ycsb(char* tracepath);
void exeWritetrace(WriteRequest* request);
void exeReadtrace(ReadRequest* request);
void send_WriteRequest(WriteRequest* request,int fd);
void send_ReadRequest(ReadRequest* request,int fd);
void recv_WriteResponse(WriteResponse* response,int fd);
void recv_ReadResponse(ReadResponse* response,int fd);
void init_block_loc();
void reconstruction(int recovery_id,int blocknum);

int main(int argc, char const *argv[])
{
	if (argc!=2)
	{
		printf("please input ./client latency_file_num\n");
		return 0;
	}

	char* filename = (char*)malloc(sizeof(char)*30);
	sprintf(filename,"latency%s",argv[1]);
	latency_file = fopen(filename,"a"); 

	datanodes = read_config_Node(NODE_CONFIG);
	init_block_loc();

	for (int i = 0; i < DATANODE_NUM; ++i)
	{
		conn2client_fd[i] = connect_try(datanodes[i].tcp_ip,PORT);
		printf("connect to datanode %s:%d successfully\n",datanodes[i].tcp_ip,PORT);
	}

	sleep(10);
	printf("start playing trace\n");
	if (isYCSB==1)
	{
		play_trace_ycsb(YCSB_TRACE);
	}else{
		play_trace_msr(MSR_TRACE);
		if (IS_RECOVERY==1)
		{
			reconstruction(RECOVERY_ID,RECOVERY_NUM);	
		}
	}
	printf("finish playing trace successfully\n");
	fclose(latency_file);
	while(1){
		sleep(100);
	}

	for (int i = 0; i < DATANODE_NUM; ++i) close(conn2client_fd[i]);

	return 0;
}

void play_trace_ycsb(char* tracepath)
{
	FILE* fp;
	if ((fp=fopen(tracepath,"r"))==NULL)
	{
		printf("open trace failed in play_trace_ycsb\n");
		exit(0);
	}

	int ret;
	char operation[150];
	char blockid[20];
	char op_type[4];
	char divider=',';
	long write_count=0;
	long read_count=0;
	int blockid_int=0;

	while(fgets(operation,sizeof(operation),fp)){
		new_strtok(operation,divider,op_type);
		new_strtok(operation,divider,blockid);

		sscanf(blockid,"%d",&blockid_int);

		if ((ret = strcmp("U",op_type))==0)
		{
			write_count++;
			if (write_count%500==0)
			{
				printf("%ld write request finish \n",write_count);
			}
			start_time = getcurrent_time();

			int stripeid = blockid_int/K;
			int offsetid = blockid_int%K;
			int targetid = block_loc[stripeid][offsetid];
			WriteRequest* request = (WriteRequest*)malloc(sizeof(WriteRequest));
			request->type=1;
			request->requestid=++requestid;
			request->targetid=targetid;
			for (int j = 0; j < R; j++) {
				request->paritynodeid[j] = block_loc[stripeid][K+j];
				request->parityid[j] = stripeid*(K+R)+K+j;
			}
			request->blockid=stripeid*(K+R)+offsetid;
			gene_radm_buff(request->newdata,BLK_SIZE);
			exeWritetrace(request);
			free(request);

			end_time = getcurrent_time();
			s_time = end_time - start_time;
			printf("%ld write request latency:%ld mius\n",write_count,s_time);
			fprintf(latency_file, "%s,%ld\n",op_type,s_time);
			write_total_time=write_total_time+s_time;
			total_time=total_time+s_time;

		}else if ((ret=strcmp("R",op_type))==0){
			read_count++;
			if (read_count%500==0)
			{
				printf("%ld read request finish \n",read_count);
			}
			start_time = getcurrent_time();

			int stripeid = blockid_int/K;
			int offsetid = blockid_int%K;
			int targetid = block_loc[stripeid][offsetid];
			ReadRequest* request = (ReadRequest*)malloc(sizeof(ReadRequest));
			request->type=0;
			request->requestid=++requestid;
			request->targetid=targetid;
			request->blockid = stripeid*(K+R)+offsetid;
			exeReadtrace(request);
			free(request);

			end_time = getcurrent_time();
			s_time = end_time - start_time;
			printf("%ld read request latency:%ld mius\n",read_count,s_time);
			fprintf(latency_file, "%s,%ld\n",op_type,s_time);
			read_total_time=read_total_time+s_time;
			total_time=total_time+s_time;

		}else if((ret=strcmp("W",op_type))==0){
			printf("Write is not support now op_type: %s\n",op_type);
			continue;
		}else if((ret=strcmp("D",op_type))==0){
			printf("Delete is not support now op_type: %s\n",op_type);
			continue;
		}else{
			printf("unknown op_type: %s\n",op_type);
			continue;
		}
	}

	printf("write_count:%ld\n",write_count);
	float write_latency = write_total_time*1.0/write_count;
	printf("write_latency:%f mius\n",write_latency);

	printf("read_count:%ld\n",read_count);
	float read_latency = read_total_time*1.0/read_count;
	printf("read_latency:%f mius\n",read_latency);

	printf("total_count:%ld\n",read_count+write_count);
	float latency = total_time*1.0/(read_count+write_count);
	printf("latency:%f mius\n",latency);
	fclose(fp);
}

void play_trace_msr(char* tracepath)
{
	FILE* fp;
	if ((fp=fopen(tracepath,"r"))==NULL)
	{
		printf("open trace failed in play_trace_msr\n");
		exit(0);
	}

	char operation[150];
	char time_stamp[50];
	char workload_name[10];
	char volumn_id[5];
	char op_type[10];
	char offset[20];
	char size[10];
	char usertime[10];
	char divider=',';

	int i;
	int access_start_block,access_end_block;
	int ret;
	long write_count;
	long read_count;

	long long *size_int;
	long long *offset_int;
	long long a,b;
	a=0LL;
	b=0LL;
	offset_int=&a;
	size_int=&b;
	write_count=0;
  	read_count=0;

	while(fgets(operation,sizeof(operation),fp)){

		if(write_count+read_count>100000&&read_count>50000){
			break;
		}
		new_strtok(operation,divider,time_stamp);
		new_strtok(operation,divider,workload_name);
		new_strtok(operation,divider,volumn_id);
		new_strtok(operation,divider,op_type);
		new_strtok(operation,divider,offset);
		new_strtok(operation,divider,size);
		new_strtok(operation,divider,usertime);

		if ((ret=strcmp(op_type,"Read"))==0)
		{
			read_count++;
			if (read_count%500==0)
			{
				printf("%ld read request finish \n",read_count);
			}

			trnsfm_char_to_int(offset,offset_int);
			trnsfm_char_to_int(size,size_int);
			access_start_block=(*offset_int)/((long long)(DUMMY_BLK_SIZE));
			access_end_block=(*offset_int+*size_int-1)/((long long)(DUMMY_BLK_SIZE));

			start_time = getcurrent_time();
			for (i = access_start_block; i <= access_end_block; ++i)
			{
				//发送write请求给相应的datanode
				int stripeid = i/K;
				int offsetid = i%K;
				int targetid = block_loc[stripeid][offsetid];
				//发送读数据块请求
				ReadRequest* request = (ReadRequest*)malloc(sizeof(ReadRequest));
				request->type=0;
				request->requestid=++requestid;
				request->targetid=targetid;
				request->blockid = stripeid*(K+R)+offsetid;
				exeReadtrace(request);
				free(request);	
			}
			end_time = getcurrent_time();
			s_time = end_time - start_time;
			printf("%ld read request latency:%ld mius\n",read_count,s_time);
			fprintf(latency_file, "%s,%ld\n",op_type,s_time);
			read_total_time=read_total_time+s_time;
			total_time=total_time+s_time;
		}else if ((ret=strcmp(op_type,"Write"))==0)
		{
			write_count++;
			if (write_count%500==0)
			{
				printf("%ld write request finish \n",write_count);
			}

			trnsfm_char_to_int(offset,offset_int);
			trnsfm_char_to_int(size,size_int);
	    	//printf("111\n");
			access_start_block=(*offset_int)/((long long)(DUMMY_BLK_SIZE));
			access_end_block=(*offset_int+*size_int-1)/((long long)(DUMMY_BLK_SIZE));

			start_time = getcurrent_time();
			for (i = access_start_block; i <= access_end_block; ++i)
			{
				//发送write请求给相应的datanode
				int stripeid = i/K;
				int offsetid = i%K;
				int targetid = block_loc[stripeid][offsetid];
				WriteRequest* request = (WriteRequest*)malloc(sizeof(WriteRequest));
				request->type=1;
				request->requestid=++requestid;
				request->targetid=targetid;
				for (int j = 0; j < R; j++) {
					request->paritynodeid[j] = block_loc[stripeid][K+j];
					request->parityid[j] = stripeid*(K+R)+K+j;
				}
				request->blockid=stripeid*(K+R)+offsetid;
	      		//printf("targetid:%d,paritynodeid1:%d,paritynodeid2:%d\n",request->targetid,request->paritynodeid[0],request->paritynodeid[1]);
				gene_radm_buff(request->newdata,BLK_SIZE);
				exeWritetrace(request);
				free(request);
			}
			end_time = getcurrent_time();
			s_time = end_time - start_time;
			printf("%ld write request latency:%ld mius\n",write_count,s_time);
			fprintf(latency_file, "%s,%ld\n",op_type,s_time);
			write_total_time=write_total_time+s_time;
			total_time=total_time+s_time;
		}else{
			printf("unknown op_type: %s\n",op_type);
			continue;
		}		
	}

	printf("write_count:%ld\n",write_count);
	//printf("write_total_time:%ld\n",write_total_time);
	//float write_thoughtput = write_count*1.0/write_total_time*1000*1000;
	//printf("write_thoughtput:%f ops\n",write_thoughtput);
	float write_latency = write_total_time*1.0/write_count;
	printf("write_latency:%f mius\n",write_latency);

	printf("read_count:%ld\n",read_count);
	//printf("read_total_time:%ld\n",read_total_time);
	//float read_thoughtput = read_count*1.0/read_total_time*1000*1000;
	//printf("read_thoughtput:%f ops\n",read_thoughtput);
	float read_latency = read_total_time*1.0/read_count;
	printf("read_latency:%f mius\n",read_latency);

	printf("total_count:%ld\n",read_count+write_count);
	//printf("total_time:%ld\n",total_time);
	//float thoughtput = (read_count+write_count)*1.0/total_time*1000*1000;
	//printf("thoughtput:%f ops\n",thoughtput);
	float latency = total_time*1.0/(read_count+write_count);
	printf("latency:%f mius\n",latency);
}

void reconstruction(int recovery_id,int blocknum){
	printf("start reconstruction\n");
	long time1,time2;
	time1 = getcurrent_time();
	ReadRequest* request = (ReadRequest*)malloc(sizeof(ReadRequest));
	char** blocks = (char**)malloc(sizeof(char*)*K);
	for (int i = 0; i < K; ++i)
	{
		blocks[i] = (char*)malloc(sizeof(char)*BLK_SIZE);
	}

	for (int i = 0; i < blocknum; ++i)
	{
		int count =0;
		for (int j = 0; j < DATANODE_NUM; ++j)
		{
			if (j!=recovery_id){
				request->type=0;
				request->requestid=0;
				request->targetid=j;
				request->blockid = 1;
				exeReadtrace(request);
				count++;
			}	
			if (count==K) break;
		}
		for (int j = 0; j < K; ++j)
		{
			for (int k = 0; k < BLK_SIZE; ++k)
			{
				blocks[j][k] = blocks[j][k]^0;
			}
		}
	}
	time2 = getcurrent_time();
	printf("reconstruction finish\n");
	printf("reconstruction time:%ld mius\n",time2-time1);
}

void exeReadtrace(ReadRequest* request)
{
	printf("start read_request %ld\n",request->requestid);
	int fd  = conn2client_fd[request->targetid];
	send_ReadRequest(request,fd);
	//wait response
	ReadResponse* response  = (ReadResponse*)malloc(sizeof(ReadResponse));
	recv_ReadResponse(response,fd);
	printf("finish read_request %ld\n",request->requestid);
	free(response);
}

void exeWritetrace(WriteRequest* request)
{
	//send request
	printf("send request to datanode %d\n",request->targetid);
	int fd  = conn2client_fd[request->targetid];
	send_WriteRequest(request,fd);
	//wait response
	WriteResponse* response  = (WriteResponse*)malloc(sizeof(WriteResponse));
	recv_WriteResponse(response,fd);
	printf("recv response from datanode %d\n", request->targetid);
	free(response);
}

void send_WriteRequest(WriteRequest* request,int fd)
{
	size_t needSend =sizeof(WriteRequest);
    char *sendBuf=(char*)malloc(needSend);
    memcpy(sendBuf,request,needSend); 
    send_bytes(fd,sendBuf,needSend);
    free(sendBuf);
}

void send_ReadRequest(ReadRequest* request,int fd)
{
	size_t needSend =sizeof(ReadRequest);
    char *sendBuf=(char*)malloc(needSend);
    memcpy(sendBuf,request,needSend); 
    send_bytes(fd,sendBuf,needSend);
    free(sendBuf);
}

void recv_WriteResponse(WriteResponse* response,int fd)
{
	size_t needRecv =sizeof(WriteResponse);
    char *recvBuf=(char *)malloc(needRecv);
    recv_bytes(fd,recvBuf,needRecv);
    memcpy(response,recvBuf,needRecv);  //反序列化
    free(recvBuf);
}

void recv_ReadResponse(ReadResponse* response,int fd)
{
	size_t needRecv =sizeof(ReadResponse);
    char *recvBuf=(char *)malloc(needRecv);
    recv_bytes(fd,recvBuf,needRecv);
    memcpy(response,recvBuf,needRecv);  //反序列化
    free(recvBuf);
}

void init_block_loc()
{
	block_loc = (int**)malloc(sizeof(int*)*STRIPE_NUM);
	for (int i = 0; i < STRIPE_NUM; ++i) 
		block_loc[i] = (int*)malloc(sizeof(int)*(K+R));
	for (int i = 0; i < STRIPE_NUM; ++i)
	{
		int start = i%DATANODE_NUM;
		for (int j = 0; j < K+R; ++j)
		{
			int index = (start+j)%DATANODE_NUM;
			block_loc[i][j] = index; 
		}
	}
}
